package org.tao.lightningmq.broker.event.spi.listenr;

import com.alibaba.fastjson2.JSON;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tao.lightningmq.broker.cache.CommonCache;
import org.tao.lightningmq.broker.event.model.ConsumeMsgRetryEvent;
import org.tao.lightningmq.broker.model.ConsumeQueueConsumeAckModel;
import org.tao.lightningmq.broker.model.MqTopicModel;
import org.tao.lightningmq.broker.rebalance.ConsumerInstance;
import org.tao.lightningmq.broker.timewheel.DelayMessageDTO;
import org.tao.lightningmq.broker.timewheel.SlotStoreTypeEnum;
import org.tao.lightningmq.common.dto.*;
import org.tao.lightningmq.common.enums.AckStatus;
import org.tao.lightningmq.common.enums.BrokerResponseCode;
import org.tao.lightningmq.common.event.Listener;
import org.tao.lightningmq.common.tcp.TcpMsg;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.Arrays;
import java.util.List;
import java.util.Map;

/**
 * @Author lht
 * @date 2025/1/22 - 15:34
 * @description:
 */
public class ConsumeMsgRetryListener implements Listener<ConsumeMsgRetryEvent> {

    private static final Logger LOGGER = LoggerFactory.getLogger(ConsumeMsgRetryListener.class);

    private static final int[] RETRY_STEP = new int[]{3, 5, 10, 15, 30};

    @Override
    public void onReceive(ConsumeMsgRetryEvent event) {
        ConsumeMsgRetryReqDTO consumeMsgRetryReqDTO = event.getConsumeMsgRetryReqDTO();
        ConsumeMsgRetryRespDTO consumeMsgRetryRespDTO = new ConsumeMsgRetryRespDTO();
        consumeMsgRetryRespDTO.setMsgId(event.getMsgId());
        InetSocketAddress inetSocketAddress = (InetSocketAddress) event.getContext().channel().remoteAddress();
        for (ConsumeMsgRetryReqDetailDTO consumeMsgRetryReqDetailDTO : consumeMsgRetryReqDTO.getConsumeMsgRetryReqDetailDTOList()) {
            //如果参数异常，中间会抛出异常，不会继续后续的ack和重新发送topic
            consumeMsgRetryReqDetailDTO.setIp(inetSocketAddress.getHostString());
            consumeMsgRetryReqDetailDTO.setPort(inetSocketAddress.getPort());
            this.checkParam(consumeMsgRetryRespDTO, event, consumeMsgRetryReqDetailDTO);
        }
        for (ConsumeMsgRetryReqDetailDTO consumeMsgRetryReqDetailDTO : consumeMsgRetryReqDTO.getConsumeMsgRetryReqDetailDTOList()) {
            this.ackAndSendToRetryTopic(consumeMsgRetryReqDetailDTO, event);
        }
        consumeMsgRetryRespDTO.setAckStatus(AckStatus.SUCCESS.getCode());
        TcpMsg retryMsg = new TcpMsg(BrokerResponseCode.CONSUME_MSG_RETRY_RESP.getCode(), JSON.toJSONBytes(consumeMsgRetryRespDTO));
        event.getContext().writeAndFlush(retryMsg);
    }

    private void ackAndSendToRetryTopic(ConsumeMsgRetryReqDetailDTO consumeMsgRetryReqDetailDTO, ConsumeMsgRetryEvent event) {
        String topic = consumeMsgRetryReqDetailDTO.getTopic();
        LOGGER.info(topic);
        Integer queueId = consumeMsgRetryReqDetailDTO.getQueueId();
        String consumerGroup = consumeMsgRetryReqDetailDTO.getConsumerGroup();
        // 保障后续消息可以被正常消费，避免堆积，随后把这条消息重新发送到重试topic
        CommonCache.getConsumeQueueConsumeHandler().ack(new ConsumeQueueConsumeAckModel(
                topic, consumerGroup, queueId, 1
        ));
        Integer commitLogMsgLength = consumeMsgRetryReqDetailDTO.getCommitLogMsgLength();
        Long commitLogOffset = consumeMsgRetryReqDetailDTO.getCommitLogOffset();
        MessageRetryDTO messageRetryDTO = new MessageRetryDTO();
        messageRetryDTO.setTopic(topic);
        messageRetryDTO.setQueueId(consumeMsgRetryReqDetailDTO.getQueueId());
        messageRetryDTO.setConsumeGroup(consumeMsgRetryReqDetailDTO.getConsumerGroup());
        messageRetryDTO.setSourceCommitLogOffset(Math.toIntExact(commitLogOffset));
        messageRetryDTO.setSourceCommitLogSize(commitLogMsgLength);
        messageRetryDTO.setCurrentRetryTimes(consumeMsgRetryReqDetailDTO.getRetryTime());
        LOGGER.info("consumeMsgRetryReqDetailDTO.getRetryTime(): {}", consumeMsgRetryReqDetailDTO.getRetryTime());

        int nextRetryTimeStep = 0;
        if (consumeMsgRetryReqDetailDTO.getRetryTime() >= RETRY_STEP.length) {
            MessageDTO messageDTO = new MessageDTO();
            messageDTO.setMsgId(event.getMsgId());
            messageDTO.setRetry(true);
            messageDTO.setBody(JSON.toJSONBytes(messageRetryDTO));
            messageDTO.setTopic("dead_queue");
            try {
                CommonCache.getCommitLogAppendHandler().appendMsg(messageDTO);
                LOGGER.info("将消息投递到死信队列");
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
            return;
        } else {
            long nextRetryTime = System.currentTimeMillis() + (RETRY_STEP[consumeMsgRetryReqDetailDTO.getRetryTime()] * 1000L);
//            messageDTO.setTopic("retry%" + consumerGroup);
            messageRetryDTO.setNextRetryTime(nextRetryTime);
            nextRetryTimeStep = RETRY_STEP[consumeMsgRetryReqDetailDTO.getRetryTime()];
        }


        try {
            // 把重试消息写入到commitLog文件中
//            CommonCache.getCommitLogAppendHandler().appendMsg(messageDTO);
            // 把消息放到时间轮种，等待下次重试
            DelayMessageDTO delayMessageDTO = new DelayMessageDTO();
            delayMessageDTO.setData(messageRetryDTO);
            delayMessageDTO.setSlotStoreType(SlotStoreTypeEnum.MESSAGE_RETRY_DTO);
            delayMessageDTO.setDelay(nextRetryTimeStep);
            delayMessageDTO.setNextExecuteTime(System.currentTimeMillis() + nextRetryTimeStep * 1000);
            CommonCache.getTimeWheelModelManager().addDelayMessage(delayMessageDTO);
        } catch (Exception e) {
            LOGGER.warn("appendMsg error", e);
        }
    }

    private void checkParam(ConsumeMsgRetryRespDTO consumeMsgRetryRespDTO, ConsumeMsgRetryEvent event, ConsumeMsgRetryReqDetailDTO consumeMsgRetryReqDetailDTO) {
        String topic = consumeMsgRetryReqDetailDTO.getTopic();
        String consumeGroup = consumeMsgRetryReqDetailDTO.getConsumerGroup();
        MqTopicModel mqTopicModel = CommonCache.getMqTopicModelMap().get(topic);
        if (mqTopicModel == null) {
            //topic不存在，ack失败
            consumeMsgRetryRespDTO.setAckStatus(AckStatus.FAIL.getCode());
            event.getContext().writeAndFlush(new TcpMsg(BrokerResponseCode.CONSUME_MSG_RETRY_RESP.getCode(),
                    JSON.toJSONBytes(consumeMsgRetryRespDTO)));
            throw new RuntimeException("checkParam error");
        }
        Map<String, List<ConsumerInstance>> consumerInstanceMap = CommonCache.getConsumeHoldMap().get(topic);
        if (consumerInstanceMap == null || consumerInstanceMap.isEmpty()) {
            consumeMsgRetryRespDTO.setAckStatus(AckStatus.FAIL.getCode());
            event.getContext().writeAndFlush(new TcpMsg(BrokerResponseCode.BROKER_UPDATE_CONSUME_OFFSET_RESP.getCode(),
                    JSON.toJSONBytes(consumeMsgRetryRespDTO)));
            throw new RuntimeException("checkParam error");
        }
        List<ConsumerInstance> consumeGroupInstances = consumerInstanceMap.get(consumeGroup);
        if (consumeGroupInstances == null || consumeGroupInstances.isEmpty()) {
            consumeMsgRetryRespDTO.setAckStatus(AckStatus.FAIL.getCode());
            event.getContext().writeAndFlush(new TcpMsg(BrokerResponseCode.BROKER_UPDATE_CONSUME_OFFSET_RESP.getCode(),
                    JSON.toJSONBytes(consumeMsgRetryRespDTO)));
            throw new RuntimeException("checkParam error");
        }
        // 找到当前消费者
        String currentConsumeReqId = consumeMsgRetryReqDetailDTO.getIp() + ":" + consumeMsgRetryReqDetailDTO.getPort();
        ConsumerInstance matchInstance = consumeGroupInstances.stream().filter(item -> {
            return item.getConsumerReqId().equals(currentConsumeReqId);
        }).findAny().orElse(null);
        if (matchInstance == null) {
            consumeMsgRetryRespDTO.setAckStatus(AckStatus.FAIL.getCode());
            event.getContext().writeAndFlush(new TcpMsg(BrokerResponseCode.BROKER_UPDATE_CONSUME_OFFSET_RESP.getCode(),
                    JSON.toJSONBytes(consumeMsgRetryRespDTO)));
            throw new RuntimeException("checkParam error");
        }
    }
}
